/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.amit.concurrency;

import java.util.ArrayList;
import org.amit.util.Concurrency;

/**
 *
 * @author amit
 */
public abstract class Operator<T, V> {

    Channel<T, V> channel;
    int numThreads;

    public Operator(Channel<T, V> channel, int numThreads) {
        this.channel = channel;
        this.numThreads = numThreads;
    }

    public Channel<T, V> getChannel() {
        return channel;
    }
    public abstract void doThis(T operand);

    protected void runNext() {
        T operand = channel.get();
        if (operand != null) {
            doThis(operand);
        }else{
            Thread.yield();
        }
    }

    protected void addToOutput(V toAdd) {
        channel.put(toAdd);
    }

    public static void main(String[] args) {
        Channel<Integer, Integer> sourceChannel = new IntegerSequenceChannel<Integer>("integerChannel", 0, 500, 1);
        Channel<Integer, String> nextChannel = new Channel<Integer, String>("finalChannel",sourceChannel);
        Operator<Integer, Integer> addOperator = new Operator<Integer, Integer>(sourceChannel, 10) {

            @Override
            public void doThis(Integer operand) {
                addToOutput(operand * 2);
            }
        };
        Operator<Integer, String> secondOperator = new Operator<Integer, String>(nextChannel, 10) {

            @Override
            public void doThis(Integer operand) {
                addToOutput(Integer.toString(4 * operand));
            }
        };
        ArrayList<Operator> operators = new ArrayList<Operator>();
        operators.add(addOperator);
        operators.add(secondOperator);
        Manager manager = new Manager(nextChannel, operators);
        manager.startThreads();
        while (!manager.checkDone()) {
            Concurrency.sleep(500);
        }
        manager.stopThreads();
        String output = null;
        while ((output = nextChannel.output.poll()) != null) {
            System.out.println(output);
        }
    }
}
